package com.zetyun.tiger.connector.cnt;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.SneakyThrows;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.util.Set;

import static com.zetyun.tiger.connector.op.v2.RedisOpV2Action.build;


@ActionMeta(
        id = "tiger.count-cust-cnt",
        tags = {"redis"},
        category = "map",
        name = "CountCustCnt",
        description = "统计客户经理、分支机构今日客户数",
        parameterClass = CountCustCntParam.class,
        status = ActionMeta.Status.PUBLIC
)
public class CountCustCntAction extends MapAction {


    // 公共揽存客户关系缓存
    private SimpleRedisContainer pbRelationshipDb;

    // 客户经理客户关系缓存
    private SimpleRedisContainer conntrRelationshipDb;

    // 实时数据库
    private SimpleRedisContainer rtDb;

    // 公共揽存客户关系缓存前缀
    private String pbRelationshipPrefix;

    // 客户经理客户关系缓存前缀
    private String conntrRelationshipPrefix;

    // 客户经理今日客户数
    private String cbConntrCnt;

    // 机构今日客户数
    private String cbOrgCnt;

    // 叶子机构今日开户客户号缓存
    private String orgOpacCustPrefix;

    // 叶子机构今日销户客户号缓存
    private String orgCacctCustPrefix;

    // 客户经理今日开户客户号缓存
    private String conntrOpacCustPrefix;

    // 客户经理今日销户客户号缓存
    private String conntrCacctCustPrefix;

    private CountCustCntParam param;

    private ObjectMapper om;

    private final Integer ttl = 60 * 60 * 24;  // 一天过期时间，仅 rtDb 需要设置

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        om = new ObjectMapper();
        param = context.getParameters();
        pbRelationshipDb = buildContainer(param.getPbRelationshipDb());
        conntrRelationshipDb = buildContainer(param.getConntrRelationshipDb());
        rtDb = buildContainer(param.getRtDb());
        pbRelationshipPrefix = param.getPbRelationshipPrefix();

        conntrRelationshipPrefix = param.getConntrRelationshipPrefix();

//        orgCntPrefix = param.getOrgCntPrefix();
        orgOpacCustPrefix = param.getOrgOpacCustPrefix();
        orgCacctCustPrefix = param.getOrgCacctCustPrefix();
        conntrOpacCustPrefix = param.getConntrOpacCustPrefix();
        conntrCacctCustPrefix = param.getConntrCacctCustPrefix();
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {


        String dayId = in.getField(param.getDayIdField(), String.class);
        String reaordOrgNo = in.getField(param.getRecordOrgNoField(), String.class);

        String conntrNo = in.getField(param.getConntrNoField(), String.class);
        String custNo = in.getField(param.getCustNoField(), String.class);
        String acctNo = in.getField(param.getAcctNoField(), String.class);
        String pdFlag = in.getField(param.getPdFlagField(), String.class);
        String opacType = in.getField(param.getOpacFlagField(), String.class);

        // 客户经理客户关系缓存
        String conntrRelationshipCacheName = conntrRelationshipPrefix.concat(conntrNo);

        // 公共揽存缓存客户号关系
        String pbRelationshipCacheName = pbRelationshipPrefix.concat(reaordOrgNo);


        String baseName = dayId.concat(":").concat(reaordOrgNo);

        // 机构今日开户客户号缓存
        String orgOpacCustCacheName = orgOpacCustPrefix.concat(baseName);

        // 机构今日销户客户号缓存
        String orgCacctCustCacheName = orgCacctCustPrefix.concat(baseName);

        // 客户经理今日开户客户号缓存
        String conntrOpacCustCacheName = conntrOpacCustPrefix.concat(baseName);

        // 客户经理今日销户客户号缓存
        String conntrCacctCustCacheName = conntrCacctCustPrefix.concat(baseName);

        // 账号列表： 默认无账号
        Set<String> allAcctNos = Sets.newHashSet();

        switch (opacType) {
            case "01":
                // 开户
                // todo 机构今日开户客户号缓存加入客户号，其实只要是开户的话，缓存就需要加
                rtDb.sadd(orgOpacCustCacheName, custNo, ttl);


                if ("N".equals(pdFlag)) {
                    // 客户经理也加一下
                    rtDb.sadd(conntrOpacCustCacheName, custNo, ttl);
                    /*
                     存在客户经理，需要判断客户经理缓存中是否存在此客户号
                     */
                    String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.add(acctNo);
                    conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                } else {
                    // 公共揽存机构 客户关系新增
                    String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.add(acctNo);
                    pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                }

                break;
            case "02":
                // 销户
                // todo 机构今日销户客户号缓存加入客户号，其实只要是销户的话，缓存就需要加
                rtDb.sadd(orgCacctCustCacheName, custNo, ttl);

                if ("N".equals(pdFlag)) {
                    // 客户经理也加一下
                    rtDb.sadd(conntrCacctCustCacheName, custNo, ttl);
                     /*
                     存在客户经理，需要判断客户经理缓存中是否存在此客户号
                     */
                    String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.remove(acctNo);
                    if (allAcctNos.size() == 0) {
                        // 删除此客户缓存
                        conntrRelationshipDb.hdel(conntrRelationshipCacheName, custNo);
                    } else {
                        conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                    }
                } else {
                    // 公共揽存机构 客户关系新增
                    String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.remove(acctNo);
                }
                if (allAcctNos.size() == 0) {
                    // 删除此客户缓存
                    pbRelationshipDb.hdel(pbRelationshipCacheName, custNo);
                } else {
                    pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                }


                break;
            default:
                throw new IllegalArgumentException("不存在此类 开销户类型 {}" + opacType);
        }

        Long orgCustCnt = rtDb.scard(orgOpacCustCacheName) - rtDb.scard(orgCacctCustCacheName);
        Long conntrCustCnt = rtDb.scard(conntrOpacCustCacheName) - rtDb.scard(conntrCacctCustCacheName);

        in.setValue(param.getCbOrgCnt(), orgCustCnt);
        in.setValue(param.getCbConntrCnt(), conntrCustCnt);

        return in;
    }


    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }

    private SimpleRedisContainer buildContainer(Integer db) {
        FlinkJedisSentinelConfig pbRelationshipDbConf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(param.getMasterName())
                        .setSentinels(param.getSentinels())
                        .setPassword(param.getPassword())
                        .setDatabase(db)
                        .build();

        return build(pbRelationshipDbConf);
    }
}
